package com.chinaoryx.er;

import javax.sql.DataSource;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;

/**
 * Created by IntelliJ IDEA.
 * User: hyman.zhang
 * Date: 12-2-7
 * Time:
 */
public class TransferThreadManager {
    
    private BlockingQueue<Map<String,Object>> queue = new ArrayBlockingQueue<Map<String, Object>>(1000);
    private boolean isDone = false;

    public BlockingQueue<Map<String, Object>> getQueue() {
        return queue;
    }

    public void setQueue(BlockingQueue<Map<String, Object>> queue) {
        this.queue = queue;
    }

    public boolean isDone() {
        return isDone;
    }

    public void setDone(boolean done) {
        isDone = done;
    }

    public void main(String[] args) {

    }



    public void moveEntityList(DataSource frmDs, DataSource toDs, EntityDefination entity, int threadCount, int batchCount) throws SQLException, InterruptedException {

        Statement stmt = null;
         Connection fromConn = frmDs.getConnection();

        String sql = entity.getSqlForList();
        List<Map> list = new ArrayList<Map>();
        stmt = null;
        ResultSet rs = null;

        stmt = fromConn.createStatement();
        rs = stmt.executeQuery(sql);
        ResultSetMetaData rsmd = rs.getMetaData();
        int count = rsmd.getColumnCount();
        this.isDone = false;

        // start thread
//        int threadCount = threadCount;
        List<ThreadObject> threadList = new ArrayList<ThreadObject>();
        for (int i = 0; i < threadCount; i++) {
            ThreadObject to = new ThreadObject();
            to.frmConn = frmDs.getConnection();
            to.toConn = toDs.getConnection();
            to.frmConn.setAutoCommit(false);
            to.toConn.setAutoCommit(false);
            
            TransferThread tt = new TransferThread("thread" + i, to.frmConn,to.toConn, entity, this, batchCount);
            Thread t = new Thread(tt);
            t.start();
            to.thread = t;
            threadList.add(to);
        }


        while (rs.next()) {
            Map<String, Object> result = new HashMap<String, Object>();

            for (int i = 0; i<count; i++) {
                String columnLabel = rsmd.getColumnLabel(i+1);
                Object val = rs.getObject(i+1) ;
                result.put(columnLabel, val);
            }
            queue.put(result);
        }

        this.isDone = true;
        // waiting thread finished,
        for (ThreadObject threadObject : threadList) {
            threadObject.thread.join();
        }

        //   close connection
        for (ThreadObject threadObject : threadList) {
            threadObject.frmConn.close();
            threadObject.toConn.close();
        }

        rs.close();
        stmt.close();
        fromConn.close();



    }
    
    static class ThreadObject {
        Thread thread;
        Connection frmConn;
        Connection toConn;
    }
}
